package com.rpc.server;

import com.rpc.codec.RpcDcoder;
import com.rpc.codec.RpcEncoder;
import com.rpc.codec.RpcRequest;
import com.rpc.codec.RpcResponse;
import com.rpc.config.provider.ProviderConfig;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
import lombok.extern.slf4j.Slf4j;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeUnit;

/**
 * @Description server端
 * @Author 顾志杰
 * @Date 2022/3/3 09:04
 */
@Slf4j
public class RpcServer {

  /**
   * 服务地址
   */
  private String serverAddress;

  /**
   * 接收连接
   */
  private EventLoopGroup bossGroup = new NioEventLoopGroup();

  /**
   * 实际处理逻辑
   */
  private EventLoopGroup workerGroup = new NioEventLoopGroup();

  /**
   * key interfaceName,value 接口的实现类
   */
  private volatile Map<String,Object> handlerMap= new HashMap<String,Object>();

  public RpcServer(String serverAddress) throws InterruptedException {
    this.serverAddress = serverAddress;
    this.start();
  }

  /**
   * 服务启动方法
   * @throws InterruptedException
   */
  private void start() throws InterruptedException {
    ServerBootstrap serverBootstrap = new ServerBootstrap();
    serverBootstrap.group(bossGroup, workerGroup)
      .channel(NioServerSocketChannel.class)
      /*tcp = sync + accept = backlog*/
      .option(ChannelOption.SO_BACKLOG, 1024)
      .childHandler(new ChannelInitializer<SocketChannel>() {
        @Override
        protected void initChannel(SocketChannel ch) throws Exception {
          ChannelPipeline cp = ch.pipeline();
          cp.addLast(new LengthFieldBasedFrameDecoder(
            /*整个数据包大小*/
            65536,
            /*数据包起始位置*/
            0,
            /*第一个关键数据位置(编码的时候包头有4个字节从第5个开始是真正的数据)*/
            4,
            0,
            0
          ));
          /*编解码的handler*/
          cp.addLast(new RpcDcoder(RpcRequest.class));
          cp.addLast(new RpcEncoder(RpcResponse.class));
          /*实际的业务处理器RpcClientHandler*/
          cp.addLast(new RpcserverHandler(handlerMap));
        }
      });

    String[] array = serverAddress.split(":");
    String host = array[0];
    int port = Integer.parseInt(array[1]);

    /*异步回调*/
    ChannelFuture channelFuture = serverBootstrap.bind(host, port).sync();
    /*回调 new ChannelFutureListener*/
    channelFuture.addListener((future -> {
      if (future.isSuccess()) {
        log.info("server success bing to {}", serverAddress);
      } else {
        log.info("server fail bing to {}", serverAddress);
        throw new Exception("server start fail cause: {}",future.cause());
      }
    }));

    /*同步等待*/
    try {
      channelFuture.await(5000, TimeUnit.MILLISECONDS);
      if (channelFuture.isSuccess()) {
        log.info("start rapid rpc success!");
      }
    } catch (InterruptedException e) {
      log.error("start rapid rpc occur Interrupted,ex: " + e);
    }
  }

  /**
   * 程序注册器
   */
  public void registerProcessor(ProviderConfig providerConfig) {
    /* key 接口权限命名*/
    /* value 接口下具体实现类的实例对象*/
    handlerMap.put(providerConfig.getInterface(),providerConfig.getRef());
  }

  /**
   * 服务停止方法
   */
  public void close() {
    bossGroup.shutdownGracefully();
    workerGroup.shutdownGracefully();
  }
}
